package com.tuniu.agents.util;

import com.tuniu.agents.config.StreamCommonProcessor;
import com.tuniu.agents.config.StreamProcessor;
import reactor.core.publisher.Flux;

import java.util.function.Function;

/**
 * Stream processing utility class
 * Provides static convenient methods based on StreamCommonProcessor
 */
public class StreamProcessorUtil {

    private static final StreamCommonProcessor PROCESSOR = new StreamCommonProcessor();

    /**
     * Process the stream using custom extraction logic
     */
    public static Flux<String> processWithExtractor(Flux<String> flux,
                                                    Function<String, StreamProcessor.ExtractResult> extractor) {
        return PROCESSOR.process(flux, extractor);
    }

    /**
     * Process the stream using custom extraction logic, with control over whether to return the original stream
     */
    public static Flux<String> processWithExtractor(Flux<String> flux,
                                                    Function<String, StreamProcessor.ExtractResult> extractor,
                                                    boolean emitOriginal) {
        return PROCESSOR.process(flux, extractor, emitOriginal);
    }

    /**
     * Process the stream using custom extraction logic and process the extracted content
     */
    public static Flux<String> processWithExtractor(Flux<String> flux,
                                                    Function<String, StreamProcessor.ExtractResult> extractor,
                                                    Function<String, String> contentProcessor) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.process(flux, extractor);
    }

    /**
     * Process the stream using custom extraction logic, process the extracted content, and control whether to return the original stream
     */
    public static Flux<String> processWithExtractor(Flux<String> flux,
                                                    Function<String, StreamProcessor.ExtractResult> extractor,
                                                    Function<String, String> contentProcessor,
                                                    boolean emitOriginal) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.process(flux, extractor, emitOriginal);
    }

    /**
     * Process the stream using start and end markers
     */
    public static Flux<String> processWithStartAndEnd(Flux<String> flux, String start, String end) {
        return PROCESSOR.processWithStartAndEnd(flux, start, end);
    }

    /**
     * Process the stream using start and end markers, with control over whether to return the original stream
     */
    public static Flux<String> processWithStartAndEnd(Flux<String> flux, String start, String end,
                                                      boolean emitOriginal) {
        return PROCESSOR.processWithStartAndEnd(flux, start, end, emitOriginal);
    }

    /**
     * Process the stream using start and end markers and process the extracted content
     */
    public static Flux<String> processWithStartAndEnd(Flux<String> flux, String start, String end,
                                                      Function<String, String> contentProcessor) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithStartAndEnd(flux, start, end);
    }

    /**
     * Process the stream using start and end markers, process the extracted content, and control whether to return the original stream
     */
    public static Flux<String> processWithStartAndEnd(Flux<String> flux, String start, String end,
                                                      Function<String, String> contentProcessor,
                                                      boolean emitOriginal) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithStartAndEnd(flux, start, end, emitOriginal);
    }

    /**
     * Process based on the start marker and fixed character count
     */
    public static Flux<String> processWithCharCount(Flux<String> flux, String start, int charCount) {
        return PROCESSOR.processWithCharCount(flux, start, charCount);
    }

    /**
     * Process based on the start marker and fixed character count, with control over whether to return the original stream
     */
    public static Flux<String> processWithCharCount(Flux<String> flux, String start, int charCount,
                                                    boolean emitOriginal) {
        return PROCESSOR.processWithCharCount(flux, start, charCount, emitOriginal);
    }

    /**
     * Process based on the start marker and fixed character count, and process the extracted content
     */
    public static Flux<String> processWithCharCount(Flux<String> flux, String start, int charCount,
                                                    Function<String, String> contentProcessor) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithCharCount(flux, start, charCount);
    }

    /**
     * Process based on the start marker and fixed character count, process the extracted content, and control whether to return the original stream
     */
    public static Flux<String> processWithCharCount(Flux<String> flux, String start, int charCount,
                                                    Function<String, String> contentProcessor,
                                                    boolean emitOriginal) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithCharCount(flux, start, charCount, emitOriginal);
    }

    /**
     * Process the stream using regular expressions
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex) {
        return PROCESSOR.processWithRegex(flux, regex, groupIndex);
    }

    /**
     * Process the stream using regular expressions, with control over whether to return the original stream
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex,
                                                boolean emitOriginal) {
        return PROCESSOR.processWithRegex(flux, regex, groupIndex, emitOriginal);
    }

    /**
     * Process the stream using regular expressions and add prefixes and suffixes to matched content
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex,
                                                String prefix, String suffix) {
        return PROCESSOR.processWithRegex(flux, regex, groupIndex,
                new StreamProcessor.ContentWrapper(prefix, suffix));
    }

    /**
     * Process the stream using regular expressions, add prefixes and suffixes to matched content, and control whether to return the original stream
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex,
                                                String prefix, String suffix,
                                                boolean emitOriginal) {
        return PROCESSOR.processWithRegex(flux, regex, groupIndex,
                new StreamProcessor.ContentWrapper(prefix, suffix), emitOriginal);
    }

    /**
     * Process the stream using regular expressions and process the extracted content
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex,
                                                Function<String, String> contentProcessor) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithRegex(flux, regex, groupIndex);
    }

    /**
     * Process the stream using regular expressions, process the extracted content, and control whether to return the original stream
     */
    public static Flux<String> processWithRegex(Flux<String> flux, String regex, int groupIndex,
                                                Function<String, String> contentProcessor,
                                                boolean emitOriginal) {
        return new StreamCommonProcessor() {
            @Override
            public String processCapturedContent(String content) {
                return contentProcessor.apply(content);
            }
        }.processWithRegex(flux, regex, groupIndex, emitOriginal);
    }

    /**
     * Get extraction utility method
     */
    public static StreamProcessor.ExtractResult extractByStartAndEnd(String content, String start, String end) {
        return StreamCommonProcessor.extractByStartAndEnd(content, start, end);
    }

    public static StreamProcessor.ExtractResult extractByCharCount(String content, String start, int charCount) {
        return StreamCommonProcessor.extractByCharCount(content, start, charCount);
    }

    public static StreamProcessor.ExtractResult extractByRegex(String content, String regex, int groupIndex) {
        return StreamCommonProcessor.extractByRegex(content, regex, groupIndex);
    }
}